package net.xuele.learn.rsocket.support;

import io.rsocket.Payload;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.http.client.utils.DateUtils;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import java.util.Date;

/**
 * @author: patrick-Hikari
 * @description: 消息的发布者，方法入参为消息的订阅者
 * @date: 2021/8/18 6:31 下午
 */
public class DataPublisher implements Publisher<Payload> {

    private Subscriber<? super Payload> subscriber;

    /**
     * 使得数据可以被订阅
     *
     * 在当前这个示例中，这个方法是在requestStream之后被调用的
     *
     * @param subscriber
     */
    @Override
    public void subscribe(Subscriber<? super Payload> subscriber) {
        System.out.println("i was called... " + Thread.currentThread().getName());
        this.subscriber = subscriber;
    }

    /**
     * 发布消息
     * @param payload 数据的来源，经过一系列的流转，最终来到了onNext当中
     */
    public void publish(Payload payload) {
        if (subscriber != null) {
            System.out.println("what about me??" + DateFormatUtils.ISO_DATETIME_TIME_ZONE_FORMAT.format(new Date()));
            // 订阅者收到消息，这条消息最终会传递到上面设置的那个subscriber
            subscriber.onNext(payload);
        }
    }

    public void complete() {
        if (subscriber != null) {
            subscriber.onComplete();
        }
    }
}
